package storm;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.Map;

/**
 * @author Hanhan_Jun
 * @date 2021年12月08日 9:48
 */
public class Bolt1 extends BaseRichBolt {
//    private static final long serialVersionUID = 886149197481637894L;
    private OutputCollector collector;

    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
    }

    public void execute(Tuple input) {
        String order_id = input.getStringByField("order_id");
        String book_id = input.getStringByField("book_id");
        String book_type = input.getStringByField("book_type");
        String book_name = input.getStringByField("book_name");
        String book_press = input.getStringByField("book_press");
        Double book_price = input.getDoubleByField("book_price");
        int book_num = input.getIntegerByField("book_num");
        int id = input.getIntegerByField("id");

        collector.emit(new Values(book_name, book_num, book_press,book_type,book_price));
    }

    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("book_name", "book_num", "book_press","book_type","book_price"));
    }
}
